// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include <filesystem>
#include <memory>

#include <cxxopts.hpp>

#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/exchange/base/executor/sync_executor.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/strategy/strategy.h"

#include "appcoin2/support/feed/fastfeed_util.h"

// proto
#include "coin2/exchange/feed.pb.h"

using namespace coin2::base::config;
using namespace coin2::app;
using namespace coin2::exchange::base::symbology;
using namespace coin2::exchange::feed;
using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::symbology::IProduct;
using coin2::feed::fastfeed::FastFeedWriter;
using ::fastfeed::proto::BookBuilderParameter;
using coin2::exchange::di::ConstantInjectionBuilder;


class FastFeedReplayer : public coin2::strategy::IStrategy {
 public:
  explicit FastFeedReplayer(
      const std::string& out_dir,
      const std::string& out_machine,
      const std::string& out_recipe,
      int out_worker_id)
      : out_dir_(out_dir),
        machine_(out_machine),
        recipe_(out_recipe),
        worker_id_(out_worker_id){
  }

  void Init(::coin2::app::Driver* driver) override {
    const auto& config = driver->config().feed();

    // Warmup period should be handled by parser but ignored by strategy.
    auto interval = FromIntervalConfig(config.common().archive().interval());
    start_from_ = interval.first;
    end_ts_ = interval.second;
    auto start_time = PtimeFromTimestamp(start_from_);
    CHECK_EQ(start_time.time_of_day().total_nanoseconds(), 0) <<
        "Replayer must start from the begin of a day ";
  }

  void onTradeFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onTradeFeed(*upd_clone);
  }
  void onBookFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onBookFeed(*upd_clone);
  }
  void onLiquidationFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onLiquidationFeed(*upd_clone);
  }
  void onOpenInterestFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onOpenInterestFeed(*upd_clone);
  }
  void onIndexFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onIndexFeed(*upd_clone);
  }
  void onMarkPriceFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onMarkPriceFeed(*upd_clone);
  }
  void onFundingRateFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onFundingRateFeed(*upd_clone);
  }
  void onIvGreeksFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onIvGreeksFeed(*upd_clone);
  }
  void onNavFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onNavFeed(*upd_clone);
  }
  void onKlineFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto upd_clone = CloneUpd(upd);
    GetFastfeedConverter(upd)->onKlineFeed(*upd_clone);
  }

 private:
  std::unique_ptr<FastFeedConverter> NewFastfeedConverter(
      const MarketExchangeApi& mea) {
    // create fastfeed converter for proto save
    DriverConfig app;
    FeedSystemConfig e;
    e.set_recipe(recipe_);
    e.add_archive_recipes(recipe_);
    auto exchanges = app.mutable_feed()->mutable_exchanges();
    exchanges->clear();
    (*exchanges)[mea.String()] = e;
    app.mutable_feed()->mutable_common()->set_type(
        FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE);
    auto archive = app.mutable_feed()->mutable_common()->mutable_archive();
    archive->set_worker(worker_id_);
    archive->set_archive_type(ArchiveConfig_ArchiveType::ArchiveConfig_ArchiveType_FASTFEED);
    archive->set_machine(machine_);
    auto* interval = app.mutable_feed()->mutable_common()->mutable_archive()->
    mutable_interval();
    interval->set_start(YmdFromTimestamp(start_from_));
    interval->set_end(YmdFromTimestamp(end_ts_));

    auto ci = coin2::exchange::di::ConstantInjection(coin2::exchange::di::ConstantInjectionBuilder());
    coin2::app::ConsolidateDriverConfig(ci, &app);

    auto fastfeed_converter = std::make_unique<FastFeedConverter>(out_dir_);
    fastfeed_converter->Init(&app);
    return fastfeed_converter;
  }

  std::unique_ptr<FeedUpdate> CloneUpd(const FeedUpdate& upd) {
    auto upd_clone = std::make_unique<FeedUpdate>(upd.mea().String(), upd.mea());
    upd_clone->SetRecipe(recipe_);
    upd_clone->SetTimestamp(upd.timestamp());
    upd_clone->SetType(upd.native_symbol(), upd.product(), upd.feed_type());
    upd_clone->MutableTs().set_exchange_publish_timestamp(
        upd.ts().exchange_publish_timestamp());
    upd_clone->SetTrade(&upd.trade());
    upd_clone->SetBook(upd.GetBookBuilder());
    upd_clone->SetLiquidationOrder(&upd.liquidation_order());
    upd_clone->SetOpenInterest(&upd.open_interest());
    upd_clone->SetMarkPrice(&upd.mark_price());
    upd_clone->SetIndex(&upd.index());
    upd_clone->SetFundingRate(&upd.funding_rate());
    upd_clone->SetIvGreeks(&upd.iv_greeks());
    upd_clone->SetNav(&upd.nav());
    upd_clone->SetKlines(&upd.klines());
    return upd_clone;
  }

  FastFeedConverter* GetFastfeedConverter(const FeedUpdate& upd) {
    auto mea = upd.mea();
    auto symbol = upd.product().absolute_norm();
    auto mea_s = std::make_tuple(mea, symbol);
    if (fastfeed_converters_.count(mea_s) == 0) {
      fastfeed_converters_[mea_s] = NewFastfeedConverter(upd.mea());
    }

    return fastfeed_converters_[mea_s].get();
  }
  std::experimental::filesystem::path out_dir_;
  std::string machine_;
  std::string recipe_;
  int worker_id_;
  int64_t start_from_;
  int64_t end_ts_;
  std::map<std::tuple<MarketExchangeApi, std::string>,
           std::unique_ptr<FastFeedConverter>> fastfeed_converters_;
};

int main(int argc, char* argv[]) {
  DriverConfig app;
  cxxopts::Options opt("FastFeed Replate", "Replay fastfeed to a new recipe");
  AddDriverOptions(&opt);
  opt.add_options()
      ("date", "trading date",
          cxxopts::value<std::string>())
      ("out-dir", "fastfeed out directory",
          cxxopts::value<std::string>()->default_value(
              "/remote/iosg/coin-2/buckets/feed.derived.feed_cache.coin2"))
      ("out-machine", "fastfeed output machine",
          cxxopts::value<std::string>()->default_value(
              "feed-05.ap-northeast-1.aws"))
      ("out-recipe", "fastfeed output recipe",
          cxxopts::value<std::string>()->default_value(
              "non_tbs"))
      ("out-worker", "fastfeed output worker id",
          cxxopts::value<int>()->default_value(
              "0"));

  auto res = opt.parse(argc, argv);
  CHECK(ParseDriverOptions(res, &app)) << "Fail to parse driver";

  // port optoin date to dirver config
  if (res.count("date") != 0U) {
    auto* interval = app.mutable_feed()->mutable_common()->mutable_archive()->
        mutable_interval();
    interval->Clear();
    interval->set_start(res["date"].as<std::string>());
    int64_t end_ts = TimestampFromString(res["date"].as<std::string>()) +
        24 * 60 * 60 * 1'000'000'000LL;
    interval->set_end(YmdFromTimestamp(end_ts));
  }

  FastFeedReplayer converter(res["out-dir"].as<std::string>(),
                             res["out-machine"].as<std::string>(),
                             res["out-recipe"].as<std::string>(),
                             res["out-worker"].as<int>());

  ConstantInjectionBuilder ci_builder = ConstantInjectionBuilder();
  coin2::app::Driver driver(app, &converter, ci_builder);
  driver.Run();

  return 0;
}
